package com.atguigu.flink0922.chapter07;

import com.atguigu.flink0922.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/3/5 15:32
 */
public class Flink08_WaterMark_OutOrder_SideOutput {
    public static void main(String[] args) {
        
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        
        env.setParallelism(1);
        // 1. 水印不会后退: 时间不会倒流
        // 2. 水印: 最大时间戳-乱序程度-1ms
        // 3. 有序数据的水印: 是乱序程度 0
        // 4. 默认情况: 周期性的向数据中插入水印 200ms
        final WatermarkStrategy<WaterSensor> wms = WatermarkStrategy
            .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
            .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                @Override
                public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                    return element.getTs();
                }
            });
    
        final SingleOutputStreamOperator<String> mainStream = env
            .socketTextStream("hadoop162", 9999)
            .map(line -> {
                final String[] data = line.split(",");
                return new WaterSensor(data[0], Long.valueOf(data[1]) * 1000, Integer.valueOf(data[2]));
            })
            .assignTimestampsAndWatermarks(wms)
            .keyBy(WaterSensor::getId)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .sideOutputLateData(new OutputTag<WaterSensor>("late_date") {})
            .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
                @Override
                public void process(String key,
                                    Context context,
                                    Iterable<WaterSensor> elements,
                                    Collector<String> out) throws Exception {
                
                    int count = 0;
                    for (WaterSensor element : elements) {
                        count++;
                    }
                
                    String msg = "当前key: " + key
                        + "窗口: [" + context.window().getStart() / 1000 + "," + context.window().getEnd() / 1000 + ") 一共有 "
                        + count + "条数据 ";
                    out.collect(msg);
                
                }
            });
        mainStream.print("main");
    
        mainStream.getSideOutput(new OutputTag<WaterSensor>("late_date") {}).print("侧");
    
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
